/****************************************************************************
 *
 * Copyright (C) 2003-2008 Los Alamos National Security, LLC
 *                         Packet Analytics Corporation
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License Version 2 as
 * published by the Free Software Foundation.  You may not use, modify or
 * distribute this program under any other version of the GNU General
 * Public License.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
 *
 ****************************************************************************/
package nfse.query;

import java.sql.*;
import java.util.ArrayList;
import java.util.Properties;
import java.net.*;
import javax.net.*;

import nfse.*;
import nfse.result.*;

import java.io.*;

/*******************************************************************************
 * This is a query processor for executing NetFSE queries over a MySQL database
 * containing TippingPoint block logs.
 * 
 * @author This code is the property of Los Alamos National Laboratory and the
 *         University of California, Copyright 2005
 */
public class DynamicQueryProcessor {

    /***************************************************************************
     * The query server that instantiated this object
     */
    private QueryServer queryServer = null;

    /***************************************************************************
     * Instantiates the query processor, associating it with the query server
     * that created it.
     * 
     * @param queryServer
     *            The server creating this query processor
     */
    public DynamicQueryProcessor(QueryServer queryServer) {
        this.queryServer = queryServer;
    }

    
    /***************************************************************************
     * Given a query, executes the query for the TippingPoint data set.
     * 
     * @param query
     *            The query to execute
     * @param db
     *            The system meta-database
     * @return a zero value on success and a negative value on failure
     */
    public int process(Query query, JDBCDatabase db) {
        Socket resultSocket = null;
        ServerSocket ss = null;
        DataOutputStream resultOut = null;
        SocketResult socketResult = null;
        MarshalledResult marshalledResult = null;
        SocketFactory sf = SocketFactory.getDefault();
        int count = 0;

        try {
            MetaDatabase mdb = (MetaDatabase) db;

            int queryID = query.getID();

            System.out.println("Processing query: '" + query.toXML() + "'");
            int numSubs = query.getNumSubqueries();
            if (numSubs > 0) {
                Query[] subs = query.getSubqueries();
                for (int i = 0; i < numSubs; i++) {
                    subs[i] = subs[i].detach(this.queryServer.nextQueryID());
                    this.process(subs[i], mdb);
                }

                System.gc();
                return 0;
            }

            if (query.getNumDataTypes() == 1) {
                Data datatype = query.getData()[0];
                if (datatype.sensor == -1) {
                    Query[] queries = query.decomposeBySensor();
                    if (queries.length == 0) {

                        return 0;
                    }

                    ResultMarshall rm = Query.marshallResults(queries, NetFSE.LOCAL_IP_ADDRESS,
                            NetFSE.RESULT_SERVER_PORT, queryID);
                    this.queryServer.getResultServer().addResultMarshall(rm);
                    for (int i = 0; i < queries.length; i++) {
                        queries[i] = queries[i].detach(queryServer.nextQueryID());
                        this.process(queries[i], db);
                    }
                    return 0;
                }
            }

            try {
                socketResult = (SocketResult) query.getResult();
            } catch (ClassCastException cce) {
                marshalledResult = (MarshalledResult) query.getResult();
            }

            int type = query.getData()[0].type;
            int sensor = query.getData()[0].sensor;

            if (socketResult != null) {
                resultSocket = sf.createSocket(socketResult.host, socketResult.port);
                resultOut = new DataOutputStream(resultSocket.getOutputStream());
            } else if (marshalledResult != null) {
                resultSocket = sf.createSocket(marshalledResult.host, marshalledResult.port);
                resultOut = new DataOutputStream(resultSocket.getOutputStream());
                resultOut.writeInt(marshalledResult.queryID);
                resultOut.writeInt(NetFSE.RESULT_SERVER_FLAG_DONE);
            }
            Data data = query.getData()[0];

            Criteria[] criteria = query.getCriteria();
            String criteriaSQL = "";
            for (int y = 0; y < criteria.length; y++) {
                Criteria c = criteria[y];
                String value1 = "";
                String value2 = "";

                if (y > 0)
                    criteriaSQL += " and ";

                criteriaSQL += NetFSE.FIELD_NAMES[c.field];

                switch (c.field) {
                case NetFSE.FIELD_DST_IP:
                case NetFSE.FIELD_SRC_IP:
                    value1 = "'" + c.value1 + "'";
                    if (c.predicate == NetFSE.PREDICATE_RANGE) {
                        value2 = "'" + c.value2 + "'";
                    }
                    break;
                default:
                    value1 = c.value1;
                    value2 = c.value2;
                    break;
                }
                if (c.predicate == NetFSE.PREDICATE_RANGE) {
                    criteriaSQL += " between " + value1 + " and " + value2 + " ";
                } else {
                    criteriaSQL += NetFSE.PREDICATES[c.predicate] + " " + value1 + " ";
                }

            }

            String sql = "";
            Connection tempConnection = null;
            Statement tempStatement = null;
            ResultSet rs = null;

            String currentHost = "";
            TimestampRange[] timestamps = query.getTimestampRanges();
            for (int x = 0; x < timestamps.length; x++) {
                String timestampSQL = " and startts between '" + timestamps[x].startTS + "' and '"
                        + timestamps[x].endTS + "' ";

                ArrayList<TempTableRun> runs = mdb.getTempTableRuns(timestamps[x].startTS, timestamps[x].endTS, sensor);
                for (int y = 0; y < runs.size(); y++) {
                    TempTableRun run = runs.get(y);
                    if (!currentHost.equals(run.getHost())) {
                        if (tempConnection != null) {
                            tempStatement.close();
                            tempConnection.close();
                        }
                        try {
                            Properties props = new Properties();
                            props.setProperty("user", run.getDynamicUser());
                            props.setProperty("password", run.getDynamicPassword());
                            props.setProperty("client_encoding", "UTF-8");
                            Class.forName("org.postgresql.Driver");
                            String url = "jdbc:postgresql://" + run.getHost() + ":" + run.getDynamicPort() + "/"
                                    + NetFSE.getTypeName(type).toLowerCase();
                            tempConnection = DriverManager.getConnection(url, props);
                            tempStatement = tempConnection.createStatement();

                        } catch (Exception E) {
                            tempConnection = null;
                            E.printStackTrace();
                        }

                    }

                    if (tempConnection != null) {
                        run = runs.get(y);
                        currentHost = run.getHost();

                        int offset = 0;
                        int limit = 10000;
                        int itterations = 0;

                        do {
                            itterations++;

                            sql = "select record, record_hex from " + runs.get(y).getTable() + " where " + criteriaSQL;
                            if (run.getMinTS() != null) {
                                if ((timestamps[x].startTS.before(run.getMinTS()))
                                        && ((timestamps[x].endTS.after(run.getMaxTS())))) {
                                    // don't add timestamp sql
                                } else if (timestamps[x].startTS.before(run.getMinTS())) {
                                    sql += " and startts <= '" + timestamps[x].endTS + "' ";
                                } else if (timestamps[x].endTS.after(run.getMaxTS())) {
                                    sql += " and startts >= '" + timestamps[x].startTS + "' ";
                                } else {
                                    sql += timestampSQL;
                                }
                            } else {
                                sql += timestampSQL;
                            }
                            sql += " limit " + limit + " offset " + offset;
                            try {
                                rs = tempStatement.executeQuery(sql);
                            } catch (Exception e) {
                                e.printStackTrace();
                                resultOut.writeInt(-100);
                                resultOut.writeInt(count);
                                DataInputStream resultIn = new DataInputStream(resultSocket.getInputStream());
                                resultIn.readInt();
                                resultIn.close();
                                resultOut.close();
                                return 0;
                            }
                            while (rs.next()) {
                                offset++;

                                String line = rs.getString("record");
                                if (line.length() == 0) {
                                    line = NetFSEUtil.convertFromHex(rs.getString("record_hex"));
                                }
                                try {
                                    resultOut.writeInt(type);
                                    resultOut.writeInt(sensor);
                                    resultOut.writeInt(line.length());
                                    resultOut.writeBytes(line);
                                    count++;
                                    if (count == data.limit) {
                                        if (marshalledResult != null) {
                                            resultOut.writeInt(-100);
                                            resultOut.writeInt(count);
                                            DataInputStream resultIn = new DataInputStream(resultSocket
                                                    .getInputStream());
                                            resultIn.readInt();
                                        }
                                        if (socketResult != null) {
                                            resultOut.writeInt(-100);
                                            resultOut.writeInt(count);
                                            DataInputStream resultIn = new DataInputStream(resultSocket
                                                    .getInputStream());
                                            resultIn.readInt();
                                        }
                                        resultOut.close();
                                        System.out.println("DynamicQueryProcessor: finished query #"
                                                + marshalledResult.queryID + ", records=" + count);

                                        rs.close();
                                        tempStatement.close();
                                        tempConnection.close();

                                        try {
                                            // sock.close();
                                            if (ss != null)
                                                ss.close();
                                            if (resultSocket != null)
                                                resultSocket.close();
                                        } catch (IOException ioe) {
                                            ioe.printStackTrace();
                                        }
                                        return 0;
                                    }

                                } catch (Exception E) {
                                    line = null;
                                }

                                if (line == null)
                                    break;
                            }
                            rs.close();

                        } while (offset / (itterations * limit) == 1);
                    }
                }
            }

            if (tempStatement != null)
                tempStatement.close();
            if (tempConnection != null)
                tempConnection.close();

            if (resultOut != null) {
                try {
                    if (marshalledResult != null) {
                        resultOut.writeInt(-100);
                        resultOut.writeInt(count);
                        DataInputStream resultIn = new DataInputStream(resultSocket.getInputStream());
                        resultIn.readInt();
                    }
                    if (socketResult != null) {
                        resultOut.writeInt(-100);
                        resultOut.writeInt(count);
                        DataInputStream resultIn = new DataInputStream(resultSocket.getInputStream());
                        resultIn.readInt();
                    }
                    System.out.println("DynamicQueryProcessor: finished query #" + marshalledResult.queryID
                            + ", records=" + count);
                    resultOut.close();
                } catch (Exception E) {

                }

            }

            try {
                // sock.close();
                if (ss != null)
                    ss.close();
                if (resultSocket != null)
                    resultSocket.close();
            } catch (IOException ioe) {
                ioe.printStackTrace();
            }

        } catch (Exception E) {
            E.printStackTrace();
            if (resultOut != null) {
                try {
                    if (marshalledResult != null) {
                        resultOut.writeInt(-100);
                        resultOut.writeInt(count);
                        DataInputStream resultIn = new DataInputStream(resultSocket.getInputStream());
                        resultIn.readInt();
                        System.out.println("DynamicQueryProcessor: finished query #" + marshalledResult.queryID
                                + ", records=" + count);
                    }
                    if (socketResult != null) {
                        resultOut.writeInt(-100);
                        resultOut.writeInt(count);
                        DataInputStream resultIn = new DataInputStream(resultSocket.getInputStream());
                        resultIn.readInt();
                    }
                    resultOut.close();

                } catch (Exception E2) {
                    E2.printStackTrace();
                }
            }
            return 0;
        }
        return 0;
    }

}